package com.tensai.study.java.newconcept.typereactiveprogram.reactor.disposable;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;

import java.time.Duration;

/**
 * Disposable 测试类
 *
 * @author Tensai
 */
@Slf4j
public class DisposableTest {

	@Test
	public void disposeTest() throws InterruptedException {
		Flux<Long> longFlux = Flux.interval(Duration.ofMillis(1));
		//take方法准确获取订阅数据量
		Disposable disposable = longFlux.take(50).subscribe(x -> log.info("->{}", x));
		log.info("sleep start");
		//sleep不能停止正在推送数据中的Flux或Mono流
		//因为不在一个线程里
		Thread.sleep(100);
		log.info("sleep end");
		//彻底停止正在推送数据中的Flux或Mono流
		disposable.dispose();
		log.info("->Stop");
	}

	/**
	 * disposable 彻底阻断Publisher的subscriber
	 *
	 * @throws InterruptedException Exception
	 */
	@Test
	public void disposable() throws InterruptedException {
		Flux<Long> longFlux = Flux.interval(Duration.ofMillis(1));
		//take方法准确获取订阅数据量
		Disposable disposable = longFlux.take(10000).subscribe(x -> {
			log.info("->{}", x);
		});
		//不能停止正在推送数据中的Flux或Mono流
		Thread.sleep(100);
		//彻底停止正在推送数据中的Flux或Mono流
		disposable.dispose();
		log.info("->Stop");
	}
}